package com.clei.utils.mqtt;

import com.clei.utils.PrintUtil;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.util.Arrays;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;

/**
 * mqtt client manager
 *
 * @author Y
 * @date 2023-12-15
 **/
public class MqttManager {

    /**
     * 连接次数
     */
    private volatile int connectTimes = 0;

    /**
     * lock
     */
    private final static Object LOCK = new Object();

    /**
     * mqtt property
     */
    private final MqttProperty mqttProperty;

    /**
     * mqtt callback
     */
    private final DefaultMqttMsgCallback mqttCallback;

    /**
     * mqtt client
     */
    private MqttClient mqttClient;

    /**
     * topic map
     * topic -> qos
     */
    private final Map<String, Integer> topicMap = new ConcurrentHashMap<>(0);

    /**
     * qos
     */
    private static final int DEFAULT_QOS = 0;

    public MqttManager(MqttProperty mqttProperty, DefaultMqttMsgCallback mqttCallback) {
        this.mqttProperty = mqttProperty;
        // topic信息
        String[] topicArr = mqttProperty.getTopicArr();
        if (null != topicArr) {
            Integer[] qosArr = mqttProperty.getQosArr();
            if (null == qosArr) {
                qosArr = Stream.iterate(0, i -> DEFAULT_QOS).limit(topicArr.length).toArray(Integer[]::new);
            }
            for (int i = 0; i < qosArr.length; i++) {
                topicMap.put(topicArr[i], qosArr[i]);
            }
        }

        this.mqttCallback = mqttCallback;
        this.mqttCallback.setReconnectTask(this::connect);
        connect();
    }

    /**
     * 发送消息
     *
     * @param topic topic
     * @param msg   msg
     */
    public void publish(String topic, String msg) {
        publish(topic, msg, DEFAULT_QOS);
    }

    /**
     * 发送消息
     *
     * @param topic topic
     * @param msg   msg
     * @param qos   qos
     */
    public void publish(String topic, String msg, int qos) {
        MqttMessage message = new MqttMessage();
        message.setPayload(msg.getBytes());
        message.setQos(qos);
        try {
            mqttClient.publish(topic, message);
        } catch (Exception e) {
            PrintUtil.log("publish error topic : {}, msg : {}", topic, msg);
        }
    }

    /**
     * 添加订阅
     *
     * @param topic topic
     */
    public void subscribe(String topic) {
        subscribe(topic, DEFAULT_QOS);
    }

    /**
     * 添加订阅
     *
     * @param topic topic
     * @param qos   qos
     */
    public void subscribe(String topic, int qos) {
        try {
            mqttClient.subscribe(new String[]{topic}, new int[]{qos});
            PrintUtil.log("Subscribe success topic : {}", topic);
            // 添加到map
            topicMap.put(topic, qos);
        } catch (Exception e) {
            PrintUtil.log("subscribe error ", e);
        }
    }

    /**
     * 连接
     */
    private void connect() {
        int ct = connectTimes;
        synchronized (LOCK) {
            if (ct != connectTimes) {
                return;
            }
            try {
                String clientId = mqttProperty.getClientId();
                MqttClient client = new MqttClient(mqttProperty.getUrl(), clientId, new MemoryPersistence());
                // MQTT 连接选项
                MqttConnectOptions connOpts = new MqttConnectOptions();
                connOpts.setUserName(mqttProperty.getUsername());
                connOpts.setPassword(mqttProperty.getPassword().toCharArray());
                // 清空会话
                connOpts.setCleanSession(true);
                // 连接超时 10秒
                connOpts.setConnectionTimeout(10);
                // 设置回调
                // 有消费的话 实现MqttCallback即可
                client.setCallback(mqttCallback);
                // 建立连接
                PrintUtil.log("Connecting to url : {}", mqttProperty.getUrl());
                client.connect(connOpts);
                PrintUtil.log("Connected");
                // 订阅
                subscribe(client);

                this.mqttClient = client;
            } catch (Exception e) {
                PrintUtil.log("Connect error ", e);
                // 无限重连
                try {
                    TimeUnit.SECONDS.sleep(3L);
                } catch (Exception ee) {
                    PrintUtil.log("sleep error ", ee);
                }
                connect();
            }
            this.connectTimes++;
        }
    }

    /**
     * 订阅
     *
     * @param mqttClient mqttClient
     * @throws Exception 订阅异常
     */
    private void subscribe(MqttClient mqttClient) throws Exception {
        // 这里使用map是保证重连的时候 订阅的主题包含中途添加的
        int size = topicMap.size();
        if (0 == size) {
            return;
        }
        String[] topicArr = new String[size];
        int[] qosArr = new int[size];
        // 数组参数
        Iterator<Map.Entry<String, Integer>> iterator = topicMap.entrySet().iterator();
        int i = 0;
        while (iterator.hasNext()) {
            Map.Entry<String, Integer> entry = iterator.next();
            topicArr[i] = entry.getKey();
            qosArr[i] = entry.getValue();
            i++;
        }
        mqttClient.subscribe(topicArr, qosArr);
        PrintUtil.log("Subscribe success topics : {}", Arrays.toString(topicArr));
    }
}
